package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux;

import org.junit.Test;
import reactor.core.publisher.Flux;

/**
 * @author Tensai
 */
public class WindowTest {

    @Test
    public void test() {
        //从起始位置开始将source sequence拆分成固定大小的flux
        Flux<Flux<Integer>> window = Flux.range(1, 10)
                .window(3);
//subscribe1
        window.subscribe(integerFlux -> {
            System.out.println("+++++分隔符+++++");
            integerFlux.subscribe(System.out::println);
        });
        System.out.println("---------- 分割线1 ----------");

//subscribe2
        window.toStream().forEach(integerFlux -> {
            System.out.println("+++++分隔符+++++");
            integerFlux.subscribe(System.out::println);
        });
        System.out.println("---------- 分割线2 ----------");

    }

    /**
     * 窗口分隔
     */
    @Test
    public void window() {
        //一维Flux
        Flux<String> stringFlux1 = Flux.just("a", "b", "c", "d", "e", "f", "g", "h", "i");
        //二维Flux
        Flux<Flux<String>> stringFlux2 = stringFlux1.window(2);
        stringFlux2.count().subscribe(System.out::println);
        stringFlux2.subscribe(it -> {
            it.subscribe(System.out::println);
            System.out.println("==============");
        });

        System.out.println("++++++++++++++++++++");
        //三维Flux
        Flux<Flux<Flux<String>>> stringFlux3 = stringFlux2.window(3);
        stringFlux3.count().subscribe(System.out::println);
        stringFlux3.subscribe(it2 -> {
            it2.subscribe(it -> {
                it.subscribe(System.out::println);
                System.out.println("--------------------");
            });
            System.out.println("********************");
        });
    }
}
